package hbase.operation;

import hbase.tools.Constants;
import hbase.tools.HbaseUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.NamespaceNotFoundException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.TableDescriptor;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class NamespaceOperation {

    //初始化连接对象
    static Connection conn = null;
    static Admin admin = null;

    static{
        try {
            // 本质上也是hadoop中的Configuration对象
            // 只不过在这个对象的基础上添加了hbase的 hbase-default.xml和hbase-site.xml
            Configuration conf = HBaseConfiguration.create();

            //设置要连接的Zookeeper的地址，也可以写在 hbase-site.xml中
            conf.set(Constants.ZK_QUORUM,Constants.ZK_QUORUM_NODES);
            conf.set(Constants.ZK_CLIENT_PORT,Constants.ZK_PORT);

            //通过连接工厂创建一个连接
            conn = ConnectionFactory.createConnection(conf);
            //通过连接获取操作namespace和ddl的对象 Admin
            admin = conn.getAdmin();
        }catch (IOException e){
            e.printStackTrace();
        }
    }


    /**
     * 创建没有属性值的 namespace
     * @param nsName
     */
    public static void createNamespace(String nsName){
        createNamespace(nsName,null);
    }

    /**
     * 创建有属性值的 namespace
     * @param nsName
     * @param propertiesMap
     */
    public static void createNamespace(String nsName, Map<String,String> propertiesMap){
        try {
            //判断此数据库是否存在
            admin.getNamespaceDescriptor(nsName);
            System.out.println(nsName+" 这个数据库已经存在,无法创建");
        } catch (NamespaceNotFoundException e) {
            try {
                //创建一个 NamespaceDescriptor 对象
                NamespaceDescriptor nd = NamespaceDescriptor.create(nsName).build();
                //如果有需要添加的参数，遍历这个map,依次添加
                if(propertiesMap!=null && !propertiesMap.isEmpty()){
                    for (Map.Entry<String, String> property : propertiesMap.entrySet()) {
                        nd.setConfiguration(property.getKey(),property.getValue());
                    }
                }
                //创建数据库
                admin.createNamespace(nd);
            } catch (IOException e1) {
                e1.printStackTrace();
            } finally {
                HbaseUtils.closed(conn,admin);
            }
        } catch (IOException e){
            e.printStackTrace();
        }
    }


    /**
     * 修改 namespace
     * @param nsName
     * @param properties 如果是List，表示要删除属性；如果是Map，表示要添加或修改属性
     */
    public static void alterNamespace(String nsName,Object properties){

        try {
            //获取当前nsName的 NamespaceDescriptor 对象
            NamespaceDescriptor old_nd = admin.getNamespaceDescriptor(nsName);

            //如果参数是列表类型，表示存储的是要删除的所有key
            if(properties instanceof List){
                List<String> propList = (List<String>)properties;
                for (String propKey : propList) {
                    old_nd.removeConfiguration(propKey);
                }
            }else if(properties instanceof Map){
                //如果参数是Map类型，表示存储的是要添加或修改的所有的键值对
                Map<String,String> propMap = (Map<String, String>) properties;
                for (Map.Entry<String, String> prop : propMap.entrySet()) {
                    old_nd.setConfiguration(prop.getKey(),prop.getValue());
                }
            }
            admin.modifyNamespace(old_nd);
        } catch (NamespaceNotFoundException e){
            System.out.println(nsName+" 数据库不存在，无法修改");
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            HbaseUtils.closed(conn,admin);
        }
    }

    /**
     * 获取 namespace 中的属性值，并关闭连接
     * @param nsName
     * @param propKeys 如果数量为0,表示输出所有的属性；如果大于0,输出的是指定key的value值
     */
    public static void getNamespaceProperties(String nsName,String... propKeys){
        getNamespaceProperties(nsName,true,propKeys);
    }

    /**
     * 获取 namespace 中的属性，不关闭连接
     * @param nsName
     * @param propKeys 如果数量为0,表示输出所有的属性；如果大于0,输出的是指定key的value值
     */
    public static void getNamespaceProperties(String nsName,boolean flag,String... propKeys){
        try {
            NamespaceDescriptor old_ns = admin.getNamespaceDescriptor(nsName);

            //如果不定长参数不为空，输出指定的属性内容
            if(propKeys!=null && propKeys.length>0){
                for (String propKey : propKeys) {
                    String value = old_ns.getConfigurationValue(propKey);
                    if(value!=null && !value.equals("")){
                        System.out.println("key:"+propKey+",value:"+value);
                    }else{
                        System.out.println("当前没有 "+propKey+" 这个属性");
                    }
                }
            }else{
                //如果不定长参数为空，就输出所有的属性
                Map<String,String> propsMap = old_ns.getConfiguration();
                for (Map.Entry<String, String> property : propsMap.entrySet()) {
                    System.out.println("key:"+property.getKey()+",value:"+property.getValue());
                }
            }
        } catch (NamespaceNotFoundException e){
            System.out.println(nsName+" 数据库不存在，无法获取");
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            if(flag){
                HbaseUtils.closed(conn,admin);
            }
        }
    }

    /**
     * 获取hbase中的所有namespace
     */
    public static void getAllNamespaceByHBase(){
        try {
            //获取hbase下的所有的namespace
            NamespaceDescriptor[] all_nds = admin.listNamespaceDescriptors();
            //遍历namespace，并解析每个namespace
            for (NamespaceDescriptor nd : all_nds) {
                String nsName = nd.getName();
                System.out.println("namespace:"+nsName);
                //如果需要输出每个namespace的属性，直接调用已有方法，不要重复写
                //这里调用其他方法，注意是否关闭连接和操作对象
                getNamespaceProperties(nsName,false);
                System.out.println("========================================");
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            HbaseUtils.closed(conn,admin);
        }
    }

    /**
     * 获取某个Namespace下的所有表
     * @param nsName
     * @param flag true:查找 TableDescriptor 类型的表;false:查找 TableName
     */
    public static void getAllTablesByNamespace(String nsName,boolean flag){

        try {
            //如果是true，查找 TableDescriptor
            //TableDescriptor 类型可以获取一个表的所有信息
            if(flag){
                List<TableDescriptor> all_tds = admin.listTableDescriptorsByNamespace(nsName.getBytes());
                for (TableDescriptor td : all_tds) {
                    System.out.println("descriptor:"+td.getTableName().toString());
                }
            }else{
                //如果是false，查找 TableName
                //TableName 只能获取表名
                TableName[] all_tableNames = admin.listTableNamesByNamespace(nsName);
                for (TableName tn : all_tableNames) {
                    System.out.println("tableName:"+tn.toString());
                    //通过 TableName 也可以获取到 TableDescriptor 类型
                    TableDescriptor td = admin.getDescriptor(tn);
                }
            }
        }catch (IOException e){
            e.printStackTrace();
        }finally {
            HbaseUtils.closed(conn,admin);
        }
    }

    /**
     * 删除 namespace
     * @param nsNames
     */
    public static void dropNamespace(String... nsNames){
        try {
            //如果有参数，就删除
            if(nsNames!=null && nsNames.length>0){
                for (String nsName : nsNames) {
                    if(HbaseUtils.isNamespaceExists(admin,nsName)){
                        admin.deleteNamespace(nsName);
                    }else{
                        System.out.println(nsName+" Namespace不存在,无法删除");
                    }
                }
            }else{
                //如果没有参数，给提示即可
                System.out.println("请指定数据库");
            }
        }catch (IOException e){
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {

        Map<String,String> propsMap = new HashMap<>();
        propsMap.put("name","dick");
        propsMap.put("gender","boy");
        propsMap.put("age","20");
        propsMap.put("address","bj");

        List<String> propsList = new ArrayList<>();
        propsList.add("age");
        propsList.add("gender");

//        NamespaceOperation.createNamespace("test1",propsMap);
//        NamespaceOperation.alterNamespace("test1",propsMap);
//        NamespaceOperation.getNamespaceProperties("test1","address","gender","aaaa");
//        NamespaceOperation.getAllNamespaceByHBase();
//        NamespaceOperation.getAllTablesByNamespace("default",false);
        NamespaceOperation.dropNamespace("myHbase","test","test1");

    }

}
